package com.demo.curator;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.*;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.retry.RetryNTimes;
import org.apache.curator.retry.RetryOneTime;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.Stat;

import java.util.List;
import java.util.concurrent.TimeUnit;

public class CuratorOperator {

    public CuratorFramework client = null;
    public static final String zkServerPath = "192.168.117.131:2181,192.168.117.131:2182,192.168.117.131:2183";

    /**
     * 实例化zk客户端
     */
    public CuratorOperator() {

        /**
         * 同步创建zk实例, 原生api是异步的
         *
         * curator连接zookeeper的策略: ExponentialBackoffRetry
         * baseSleepTimeMs: 初始sleep的时间
         * maxRetries: 最大
         */
//        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 5);

        RetryPolicy retryPolicy = new RetryNTimes(3, 5000);

//        RetryPolicy retryPolicy2 = new RetryOneTime(3000);
        client = CuratorFrameworkFactory.builder()
                .connectString(zkServerPath)
                .sessionTimeoutMs(10000).retryPolicy(retryPolicy)
                .namespace("workspace").build();
        client.start();
    }

    public void closeZKClient() {
        if (client != null) {
            client.close();
        }
    }

    public static void main(String[] args) throws Exception {
        CuratorOperator cto = new CuratorOperator();
        boolean isZkCuratorStarted = cto.client.isStarted();
        System.out.println("当前客户的状态：" + (isZkCuratorStarted ? "连接中" : "已关闭"));

        // 创建节点
        String nodePath = "/super/imooc";
//        byte[] data = "superme".getBytes();
//        cto.client.create().creatingParentsIfNeeded()
//                .withMode(CreateMode.PERSISTENT)
//                .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
//                .forPath(nodePath, data);

        // 更新数据
//        byte[] newData = "batman".getBytes();
//        cto.client.setData()
//                .withVersion(0)
//                .forPath(nodePath, newData);

        // 删除节点
//        cto.client.delete()
//                .guaranteed()
//                .deletingChildrenIfNeeded()
//                .withVersion(0)
//                .forPath(nodePath);

        // 读取节点数据
//        Stat stat = new Stat();
//        byte[] data = cto.client.getData().storingStatIn(stat).forPath(nodePath);
//        System.out.println("节点" + nodePath + "的数据为：" + new String(data));
//        System.out.println("该节点的版本号为：" + stat.getVersion());

        // 查询子节点
//        List<String> childNodes = cto.client.getChildren().forPath(nodePath);
//        System.out.println("开始打印子节点");
//        for (String s : childNodes) {
//            System.out.println(s);
//        }

        // 判断节点是否存在, 如果不存在则为空
//        Stat statExit = cto.client.checkExists().forPath(nodePath + "/abc");
//        System.out.println(statExit);

        // watcher事件，当使用usingWatcher的时候，监听只会触发一次，监听完毕就销毁
//        cto.client.getData().usingWatcher(new MyCuratorWatcher()).forPath(nodePath);
//        cto.client.getData().usingWatcher(new MyWatcher()).forPath(nodePath);

        // 为节点添加watcher
        // NodeCache: 监听数据节点的变更, 会触发事件
//        final NodeCache nodeCache = new NodeCache(cto.client, nodePath);
        // buildInitial: 初始化的时候获取node的值并缓存
//        nodeCache.start(true);
//        if (nodeCache.getCurrentData() != null) {
//            System.out.println("节点初始化数据为：" + new String(nodeCache.getCurrentData().getData()));
//        } else {
//            System.out.println("节点初始化数据为空...");
//        }
//        nodeCache.getListenable().addListener(new NodeCacheListener() {
//            public void nodeChanged() throws Exception {
//                String data = new String(nodeCache.getCurrentData().getData());
//                System.out.println("节点路径：" + nodeCache.getCurrentData().getPath() + "数据：" + data);
//            }
//        });

        // 为子节点添加watcher
        // PathChildrenCache: 监听数据节点的增删改，会触发事件
        String childNodePathCache = nodePath;
        final PathChildrenCache childrenCache = new PathChildrenCache(cto.client, childNodePathCache, true);
        childrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);

        List<ChildData> childDataList = childrenCache.getCurrentData();
        System.out.println("当前数据节点的子节点列表：");
        for (ChildData childData : childDataList) {
            String data = new String(childData.getData());
            System.out.println(data);
        }

        childrenCache.getListenable().addListener(new PathChildrenCacheListener() {
            public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent pathChildrenCacheEvent) throws Exception {
                if (pathChildrenCacheEvent.getType().equals(PathChildrenCacheEvent.Type.INITIALIZED)) {
                    System.out.println("子节点初始化ok...");
                } else if (pathChildrenCacheEvent.getType().equals(PathChildrenCacheEvent.Type.CHILD_ADDED)) {
                    System.out.println("添加子节点：" + pathChildrenCacheEvent.getData().getPath());
                    System.out.println("子节点数据：" + new String(pathChildrenCacheEvent.getData().getData()));
                } else if (pathChildrenCacheEvent.getType().equals(PathChildrenCacheEvent.Type.CHILD_REMOVED)) {
                    System.out.println("删除子节点：" + pathChildrenCacheEvent.getData().getPath());
                } else if (pathChildrenCacheEvent.getType().equals(PathChildrenCacheEvent.Type.CHILD_UPDATED)) {
                    System.out.println("修改子节点路径：" + pathChildrenCacheEvent.getData().getPath());
                    System.out.println("修改子节点数据：" + new String(pathChildrenCacheEvent.getData().getData()));
                }
            }
        });

        TimeUnit.SECONDS.sleep(1000);
        cto.closeZKClient();
        boolean isZkCuratorStarted2 = cto.client.isStarted();
        System.out.println("当前客户的状态：" + (isZkCuratorStarted2 ? "连接中" : "已关闭"));
    }
}
